package dm.ass1;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.commons.lang.ArrayUtils;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import dm.ass1.io.ProfileWritable;
import dm.ass1.io.RatingWritable;

public class ProfileReducer extends Reducer<Text, RatingWritable, Text, ProfileWritable> {

	@Override
	protected void reduce(Text key, Iterable<RatingWritable> value, Context context) throws IOException, InterruptedException {
		List<Integer> rated = new ArrayList<Integer>();
		List<Integer> rating = new ArrayList<Integer>();
		
		String keyString = key.toString(); 
		boolean user = keyString.startsWith(ProfilingDriver.PREFIX_USER);
		String rater = keyString.substring(1);
		
		for (RatingWritable ratingWritable : value) {
			if (user) { // This is a user profile so we populate rated with items
				rated.add(ratingWritable.getItem());
			} else { // This is an item profile so we populate rated with users
				rated.add(ratingWritable.getUser());
			}
			rating.add(ratingWritable.getRating());
		}
		
		
		int[] ratedArray = ArrayUtils.toPrimitive(rated.toArray(new Integer[rated.size()]));
		int[] ratingArray = ArrayUtils.toPrimitive(rating.toArray(new Integer[rating.size()]));
		
		ProfileWritable pw = new ProfileWritable(Integer.parseInt(rater), ratedArray, ratingArray);
		context.write(key, pw);
	}
	

}
